package com.cetc.sdp.kmga.cs.stream.nv;

import com.cetc.sdp.kmga.cs.audit.AlertReason;
import com.cetc.sdp.kmga.cs.audit.AuditRuleFastMatcher;
import com.cetc.sdp.kmga.cs.audit.AuditRuleMatchResult;
import com.cetc.sdp.kmga.cs.audit.ResourceAuditObj;
import com.cetc.sdp.kmga.cs.common.RabbitMQConf;
import com.cetc.sdp.kmga.cs.common.TableMeta;
import com.cetc.sdp.kmga.cs.jdbc.AuditDAO;
import com.cetc.sdp.kmga.cs.stream.StreamUtils;
import com.cetc.sdp.kmga.cs.util.RabbitMQConsumer;
import com.cetc.sdp.kmga.cs.util.RabbitMQProducer;
import com.cetc.sdp.kmga.cs.util.StreamJobConf;
import com.cetc.sdp.kmga.cs.util.Tool;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.mapreduce.Job;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.util.LongAccumulator;
import scala.Tuple2;

import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.*;

import static com.cetc.sdp.kmga.cs.stream.StreamUtils.*;

/**
 * @description:
 * @author： DengQiang
 * @date: 2018/3/12 15:26
 */
public class DesensTaskLogStreaming implements StreamWork.StreamJob {

    private StreamJobConf jobConf;
    private AbstractDStream dStream;

    private static final byte[] TASK_TM = Bytes.toBytes("TASK_TM");
    private static final byte[] TASK_ID = Bytes.toBytes("TASK_ID");
    private static final byte[] TASK_NAME = Bytes.toBytes("TASK_NAME");
    private static final byte[] TASK_TYPE = Bytes.toBytes("TASK_TYPE");
    private static final byte[] SRC_ADDR = Bytes.toBytes("SRC_ADDR");
    private static final byte[] SRC_PORT = Bytes.toBytes("SRC_PORT");
    private static final byte[] SRC_NAME = Bytes.toBytes("SRC_NAME");
    private static final byte[] SRC_DB_NAME = Bytes.toBytes("SRC_DB_NAME");
    private static final byte[] SRC_TAB_NAME = Bytes.toBytes("SRC_TAB_NAME");
    private static final byte[] TAB_NUM_ROWS = Bytes.toBytes("TAB_NUM_ROWS");
    private static final byte[] DESE_DETAIL = Bytes.toBytes("DESE_DETAIL");
    private static final byte[] DESE_NUM_ROW = Bytes.toBytes("DESE_NUM_ROW");
    private static final byte[] DESE_CONDT = Bytes.toBytes("DESE_CONDT");
    private static final byte[] DEST_ADDR = Bytes.toBytes("DEST_ADDR");
    private static final byte[] DEST_PORT = Bytes.toBytes("DEST_PORT");
    private static final byte[] DEST_DB_NAME = Bytes.toBytes("DEST_DB_NAME");
    private static final byte[] DEST_TAB_NAME = Bytes.toBytes("DEST_TAB_NAME");
    private static final byte[] DEST_NAME = Bytes.toBytes("DEST_NAME");
    private static final byte[] DEST_TYPE = Bytes.toBytes("DEST_TYPE");
    private static final byte[] EXTE_1 = Bytes.toBytes("EXTE_1");
    private static final byte[] EXTE_2 = Bytes.toBytes("EXTE_2");
    private static final byte[] CLIENT_IP = Bytes.toBytes("CLIENT_IP");
    private static final byte[] USR = Bytes.toBytes("USR");
    private static final byte[] SRC_RULE_ID = Bytes.toBytes("SRC_RULE_ID");
    private static final byte[] SRC_RESRC_ID = Bytes.toBytes("SRC_RESRC_ID");
    private static final byte[] SRC_AUDIT_OBJ = Bytes.toBytes("SRC_AUDIT_OBJ");
    private static final byte[] SRC_LEVEL = Bytes.toBytes("SRC_LEVEL");
    private static final byte[] SRC_OPER_TYPE = Bytes.toBytes("SRC_OPER_TYPE");
    private static final byte[] REPT_DEVICE_NUM = Bytes.toBytes("REPT_DEVICE_NUM");
    private static final byte[] CREATE_TM = Bytes.toBytes("CREATE_TM");
    private static final byte[] ACT_RESULT = Bytes.toBytes("ACT_RESULT");
    private static final byte[] SRC_TYPE = Bytes.toBytes("SRC_TYPE");

    private static final byte[] COL_FAMILY = Bytes.toBytes("C1");

    private static final int ID_IDX = 0;
    private static final int TASK_TM_IDX = 1;
    private static final int TASK_ID_IDX = 2;
    private static final int TASK_NAME_IDX = 3;
    private static final int TASK_TYPE_IDX = 4;
    private static final int SRC_ADDR_IDX = 5;
    private static final int SRC_PORT_IDX = 6;
    private static final int SRC_NAME_IDX = 7;
    private static final int SRC_DB_NAME_IDX = 8;
    private static final int SRC_TAB_NAME_IDX = 9;
    private static final int TAB_NUM_ROWS_IDX = 10;
    private static final int DESE_DETAIL_IDX = 11;
    private static final int DESE_NUM_ROW_IDX = 12;
    private static final int DESE_CONDT_IDX = 13;
    private static final int DEST_ADDR_IDX = 14;
    private static final int DEST_PORT_IDX = 15;
    private static final int DEST_DB_NAME_IDX = 16;
    private static final int DEST_TAB_NAME_IDX = 17;
    private static final int DEST_NAME_IDX = 18;
    private static final int DEST_TYPE_IDX = 19;
    private static final int EXTE_1_IDX = 20;
    private static final int EXTE_2_IDX = 21;
    private static final int CLIENT_IP_IDX = 22;
    private static final int USR_IDX = 23;
    private static final int SRC_RULE_ID_IDX = 24;
    private static final int SRC_RESRC_ID_IDX = 25;
    private static final int SRC_AUDIT_OBJ_IDX = 26;
    private static final int SRC_LEVEL_IDX = 27;
    private static final int SRC_OPER_TYPE_IDX = 28;
    private static final int REPT_DEVICE_NUM_IDX = 29;
    private static final int CREATE_TM_IDX = 30;
    private static final int ACT_RESULT_IDX = 31;
    private static final int SRC_TYPE_IDX = 32;

    private static final byte[] SRC_ADDR_INDEX_TABLE = Bytes.toBytes("KMGA:SDATA_DESE_TASK_DB_LOG_SRC_ADDR");
    private static final byte[] DEST_ADDR_INDEX_TABLE = Bytes.toBytes("KMGA:SDATA_DESE_TASK_DB_LOG_DEST_ADDR");
    private static final byte[] CLIENT_IP_INDEX_TABLE = Bytes.toBytes("KMGA:SDATA_DESE_TASK_DB_LOG_CLIENT_IP");
    private static final byte[] LOG_LEVEL_INDEX_TABLE = Bytes.toBytes("KMGA:SDATA_DESE_TASK_DB_LOG_LEVEL");

    /**
     * 是否对脱敏日志进行规则匹配
     */
    private static boolean NEED_AUDIT = false;

    public DesensTaskLogStreaming(StreamJobConf jobConf, AbstractDStream dStream) {
        this.jobConf = jobConf;
        this.dStream = dStream;
    }

    @Override
    public void processStreaming(StreamWork.Context context) {
        Job hbaseJob = context.configJob(jobConf.getHBaseJobConf());
        String logType = jobConf.getLogType().iterator().next();
        boolean saveErrorLog = jobConf.saveErrorLog();

        RabbitMQConf mqConf = createRabbitMQConf(jobConf.getRabbitMqConf());
        Broadcast<RabbitMQConf> rabbitMqConfBc = context.getSparkContext().broadcast(mqConf);

        int alertLevel = jobConf.getAlertLevel();
        TableMeta tableMeta = AuditDAO.findTableMetaByName(jobConf.getHTableName());
        long initialSeq = StreamUtils.getInitialSeqForTable(tableMeta, LocalDate.now(Tool.zoneId), false);
        byte[] DESENS_LOG_TABLE_NAME = Bytes.toBytes(jobConf.getHTableName());

        if (NEED_AUDIT) {
            RabbitMQConsumer auditMsgConsumer = new RabbitMQConsumer();
            mqConf.getNodes().forEach(auditMsgConsumer::addAddress);
            auditMsgConsumer.setQueue(mqConf.getMqNotifyQueue());
            auditMsgConsumer.setUserName(mqConf.getMqUsername());
            auditMsgConsumer.setPassword(mqConf.getMqPassword());
            auditMsgConsumer.setExchange(mqConf.getMqNotifyExchange());
            auditMsgConsumer.setXt(RabbitMQProducer.XT.FANOUT);
            auditMsgConsumer.addMsgListener(StreamUtils.RuleFastMatcherBroadcastWrapper.getInstance(context.getSparkContext()));
            auditMsgConsumer.start();

            StreamUtils.RuleFastMatcherBroadcastWrapper.getInstance(context.getSparkContext()).setInterval(jobConf.getReBroadcastInterval());
            StreamUtils.RuleFastMatcherBroadcastWrapper.getInstance(context.getSparkContext()).setUpdateByFixRate(jobConf.autoReBroadcast());
        }

        this.dStream.getDStreamAsPlaintext(context.getStreamingContext()).transform(
                stream -> stream.map(log -> {
                    //删除日志消息上多余的引号
                    if (log.startsWith("\"")) {
                        log = log.substring(1, log.length() - 1);
                    }
                    StringTokenizer tokenizer = new StringTokenizer(log, "&");
                    Map<String, String> pair = new HashMap<>(32);
                    while (tokenizer.hasMoreTokens()) {
                        String[] tmp = tokenizer.nextToken().split("=");
                        if (tmp.length == 2 && !tmp[1].isEmpty()) {
                            pair.put(tmp[0], tmp[1]);
                        }
                    }

                    Object[] data = new Object[64];
                    data[TASK_TM_IDX] = pair.get("task_time");
                    data[TASK_ID_IDX] = pair.get("task_id");
                    data[TASK_NAME_IDX] = pair.get("task_name");
                    data[TASK_TYPE_IDX] = pair.get("task_type");
                    data[SRC_ADDR_IDX] = pair.get("src_addr");
                    data[SRC_PORT_IDX] = pair.get("src_port");
                    data[SRC_NAME_IDX] = pair.get("src_name");
                    data[SRC_DB_NAME_IDX] = pair.get("src_db_name");
                    data[SRC_TAB_NAME_IDX] = pair.get("src_tab_name");
                    data[TAB_NUM_ROWS_IDX] = pair.get("tab_num_rows");
                    data[DESE_DETAIL_IDX] = pair.get("desc_detail");
                    data[DESE_NUM_ROW_IDX] = pair.get("mask_line");
                    data[DESE_CONDT_IDX] = pair.get("dese_condt");
                    data[DEST_ADDR_IDX] = pair.get("dest_addr");
                    data[DEST_PORT_IDX] = pair.get("dest_port");
                    data[DEST_DB_NAME_IDX] = pair.get("dest_db_name");
                    data[DEST_TAB_NAME_IDX] = pair.get("dest_tab_name");
                    data[DEST_NAME_IDX] = pair.get("dest_name");
                    data[DEST_TYPE_IDX] = pair.get("dest_db_type");
                    data[EXTE_1_IDX] = pair.get("exte_1");
                    data[EXTE_2_IDX] = pair.get("exte_2");
                    data[CLIENT_IP_IDX] = pair.get("src_ip");
                    data[USR_IDX] = pair.get("usr");
                    data[SRC_LEVEL_IDX] = pair.get("log_level");
                    data[ACT_RESULT_IDX] = pair.get("status");
                    data[SRC_TYPE_IDX] = pair.get("src_db_type");
                    data[62] = pair.get("rept_device_num");
                    data[63] = log;
                    return data;
                }).filter(data -> {
                    //格式异常检查
                    try {
                        data[SRC_PORT_IDX] = Integer.valueOf((String) data[SRC_PORT_IDX]);
                        String numStr = (String) data[DESE_NUM_ROW_IDX];
                        if (null != numStr && !numStr.isEmpty()) {
                            data[DESE_NUM_ROW_IDX] = Integer.valueOf((String) data[DESE_NUM_ROW_IDX]);
                        } else {
                            data[DESE_NUM_ROW_IDX] = 0;
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        if (saveErrorLog) {
                            AuditDAO.saveErrorLog(logType, (String) data[63]);
                        }
                        return false;
                    }
                    return true;
                })
        ).foreachRDD(rdd -> {
            Broadcast<AuditRuleFastMatcher> bc = StreamUtils.RuleFastMatcherBroadcastWrapper.getInstance(context.getSparkContext()).updateAndGet();
            LongAccumulator accumulator = StreamUtils.JavaIdGenerator.getInstance(context.getSparkContext(), jobConf.getHTableName(), initialSeq);
            Long seqStart = accumulator.value();
            rdd.zipWithIndex().mapPartitionsToPair(iter -> {
                List<Tuple2<ImmutableBytesWritable, Put>> res = new ArrayList<>(2048);
                StringBuilder sb = new StringBuilder();
                long size = 0;
                List<AlertReason> alertReasons = new ArrayList<>(128);
                //生成入库时间
                LocalDateTime now = LocalDateTime.now(Tool.zoneId);
                LocalDateTime createTime = now.minusNanos(now.getNano());
                String createTimeStr = Tool.DEFAULT_TIME_FORMATTER.format(createTime);
                long currentTime = StreamWork.getIdTimestamp(createTime);

                while (iter.hasNext()) {
                    Tuple2<Object[], Long> row = iter.next();
                    Object[] data = row._1();
                    ResourceAuditObj auditObj = bc.value().getResourceId((String) data[SRC_ADDR_IDX], (Integer) data[SRC_PORT_IDX], 0);
                    String timeStr = (String) data[TASK_TM_IDX];
                    LocalDateTime time = Tool.getTimeFromOrDefault(timeStr);
                    String srcDbName = (String) data[SRC_DB_NAME_IDX];
                    String op;
                    boolean opType = (srcDbName == null || srcDbName.isEmpty());
                    if (opType) {
                        op = "读";
                    } else {
                        op = "查";
                    }

                    //设置上报设备
                    data[REPT_DEVICE_NUM_IDX] = data[62];
                    //设置操作类型
                    data[SRC_OPER_TYPE_IDX] = op;
                    //设置操作结果
                    int num =  (int)data[DESE_NUM_ROW_IDX];
                    data[ACT_RESULT_IDX] = (num > 0 ? "success" : "failed");
                    //生成时间戳
                    long timestamp = StreamWork.getIdTimestamp(time);
                    String ip = (String) data[SRC_ADDR_IDX];
                    AuditRuleMatchResult armr = null;

                    //确定脱敏类型
                    String desensType = "99999";
                    switch ((String) data[TASK_TYPE_IDX]) {
                        //数据库静态脱敏
                        case "11":
                        case "13":
                            if (auditObj != null) {
                                String typeCode = auditObj.getResTypeCode();
                                if (typeCode != null && !typeCode.isEmpty()) {
                                    desensType = "00" + typeCode;
                                }
                            }
                            break;
                        //数据库动态脱敏
                        case "12":
                            if (auditObj != null) {
                                String typeCode = auditObj.getResTypeCode();
                                if (typeCode != null && !typeCode.isEmpty()) {
                                    desensType = "01" + typeCode;
                                }
                            }
                            break;
                        //文件静态脱敏
                        case "21":
                        case "23":
                            desensType = "02999";
                            break;
                        //文件动态脱敏
                        case "22":
                            desensType = "03999";
                            break;
                        default:
                            break;
                    }

                    if (auditObj != null) {
                        /**
                         * 规则匹配
                         */
                        if (NEED_AUDIT) {
                            armr = bc.getValue().match(auditObj.getResrcId(), ip, time, op);
                        }
                        data[SRC_RESRC_ID_IDX] = String.valueOf(auditObj.getResrcId());
                        data[SRC_AUDIT_OBJ_IDX] = String.valueOf(auditObj.getId());
                    } else {
                        data[SRC_RESRC_ID_IDX] = "";
                        data[SRC_AUDIT_OBJ_IDX] = "";
                    }

                    int riskLevel = 0;
                    if (armr != null) {
                        //规则匹配成功
                        data[SRC_RULE_ID_IDX] = String.valueOf(armr.getRuleId());
                        data[SRC_LEVEL_IDX] = String.valueOf(armr.getRiskLevel());
                        riskLevel = armr.getRiskLevel();
                    } else {
                        //规则匹配失败
                        data[SRC_LEVEL_IDX] = "0";
                    }

                    //生成hbase键值对
                    long seq = seqStart + row._2;
                    String id = desensType + timestamp + seq;
                    data[ID_IDX] = id;
                    size++;

                    data[CREATE_TM_IDX] = createTimeStr;
                    Put put = new Put(Bytes.toBytes(id));
                    addCol(data[TASK_TM_IDX], TASK_TM, put);
                    addCol(data[TASK_ID_IDX], TASK_ID, put);
                    addCol(data[TASK_NAME_IDX], TASK_NAME, put);
                    addCol(data[TASK_TYPE_IDX], TASK_TYPE, put);
                    addCol(data[SRC_ADDR_IDX], SRC_ADDR, put);
                    addCol(data[SRC_PORT_IDX], SRC_PORT, put);
                    addCol(data[SRC_NAME_IDX], SRC_NAME, put);
                    addCol(data[SRC_DB_NAME_IDX], SRC_DB_NAME, put);
                    addCol(data[SRC_TAB_NAME_IDX], SRC_TAB_NAME, put);
                    addCol(data[TAB_NUM_ROWS_IDX], TAB_NUM_ROWS, put);
                    addCol(data[DESE_DETAIL_IDX], DESE_DETAIL, put);
                    addCol(data[DESE_NUM_ROW_IDX], DESE_NUM_ROW, put);
                    addCol(data[DESE_CONDT_IDX], DESE_CONDT, put);
                    addCol(data[DEST_ADDR_IDX], DEST_ADDR, put);
                    addCol(data[DEST_PORT_IDX], DEST_PORT, put);
                    addCol(data[DEST_DB_NAME_IDX], DEST_DB_NAME, put);
                    addCol(data[DEST_TAB_NAME_IDX], DEST_TAB_NAME, put);
                    addCol(data[DEST_NAME_IDX], DEST_NAME, put);
                    addCol(data[DEST_TYPE_IDX], DEST_TYPE, put);
                    addCol(data[EXTE_1_IDX], EXTE_1, put);
                    addCol(data[EXTE_2_IDX], EXTE_2, put);
                    addCol(data[CLIENT_IP_IDX], CLIENT_IP, put);
                    addCol(data[USR_IDX], USR, put);
                    addCol(data[SRC_RULE_ID_IDX], SRC_RULE_ID, put);
                    addCol(data[SRC_RESRC_ID_IDX], SRC_RESRC_ID, put);
                    addCol(data[SRC_AUDIT_OBJ_IDX], SRC_AUDIT_OBJ, put);
                    addCol(data[SRC_LEVEL_IDX], SRC_LEVEL, put);
                    addCol(data[SRC_OPER_TYPE_IDX], SRC_OPER_TYPE, put);
                    addCol(data[REPT_DEVICE_NUM_IDX], REPT_DEVICE_NUM, put);
                    addCol(data[CREATE_TM_IDX], CREATE_TM, put);
                    addCol(data[ACT_RESULT_IDX], ACT_RESULT, put);
                    addCol(data[SRC_TYPE_IDX], SRC_TYPE, put);
                    res.add(new Tuple2<>(new ImmutableBytesWritable(DESENS_LOG_TABLE_NAME), put));


                    //保存SRC_ADDR索引表
                    sb.append(desensType).append(timestamp)
                            .append(data[SRC_ADDR_IDX] != null ? data[SRC_ADDR_IDX] : "");
                    StreamUtils.idAppend(sb, 39, "|");
                    sb.append(id.substring(id.length() - 3));
                    Put srcAddr = new Put(Bytes.toBytes(sb.toString()));
                    srcAddr.addColumn(INDEX_COLUMN_FAMILY_BYTES, INDEX_COLUMN_ROWKEY_BYTES, Bytes.toBytes(id));
                    srcAddr.addColumn(INDEX_COLUMN_FAMILY_BYTES, PHOENIX_APPEND_COL, PHOENIX_APPEND_COL_VALUE);
                    res.add(new Tuple2<>(new ImmutableBytesWritable(SRC_ADDR_INDEX_TABLE), srcAddr));
                    sb.setLength(0);
                    //保存DEST_ADDR
                    sb.append(desensType).append(timestamp)
                            .append(data[DEST_ADDR_IDX] != null ? data[DEST_ADDR_IDX] : "");
                    StreamUtils.idAppend(sb, 39, "|");
                    sb.append(id.substring(id.length() - 3));
                    Put destAddr = new Put(Bytes.toBytes(sb.toString()));
                    destAddr.addColumn(INDEX_COLUMN_FAMILY_BYTES, INDEX_COLUMN_ROWKEY_BYTES, Bytes.toBytes(id));
                    destAddr.addColumn(INDEX_COLUMN_FAMILY_BYTES, PHOENIX_APPEND_COL, PHOENIX_APPEND_COL_VALUE);
                    res.add(new Tuple2<>(new ImmutableBytesWritable(DEST_ADDR_INDEX_TABLE), destAddr));
                    sb.setLength(0);
                    //保存CLIENT_IP
                    sb.append(desensType).append(timestamp)
                            .append(data[CLIENT_IP_IDX] != null ? data[CLIENT_IP_IDX] : "");
                    StreamUtils.idAppend(sb, 39, "|");
                    sb.append(id.substring(id.length() - 3));
                    Put clientIP = new Put(Bytes.toBytes(sb.toString()));
                    clientIP.addColumn(INDEX_COLUMN_FAMILY_BYTES, INDEX_COLUMN_ROWKEY_BYTES, Bytes.toBytes(id));
                    clientIP.addColumn(INDEX_COLUMN_FAMILY_BYTES, PHOENIX_APPEND_COL, PHOENIX_APPEND_COL_VALUE);
                    res.add(new Tuple2<>(new ImmutableBytesWritable(CLIENT_IP_INDEX_TABLE), clientIP));
                    sb.setLength(0);
                    //特殊处理资源ID和风险等级的索引表
                    String resId = (String) data[SRC_RESRC_ID_IDX];
                    sb.append(resId == null || resId.isEmpty() ? "9999999999" : resId)
                            .append(timestamp)
                            .append(id.substring(id.length() - 8));
                    Put tmp = new Put(Bytes.toBytes(sb.toString()));
                    tmp.addColumn(INDEX_COLUMN_FAMILY_BYTES, INDEX_COLUMN_ROWKEY_BYTES, Bytes.toBytes(id));
                    tmp.addColumn(INDEX_COLUMN_FAMILY_BYTES, PHOENIX_APPEND_COL, PHOENIX_APPEND_COL_VALUE);
                    res.add(new Tuple2<>(new ImmutableBytesWritable(LOG_LEVEL_INDEX_TABLE), tmp));
                    sb.setLength(0);

                    //告警
                    if (NEED_AUDIT && riskLevel >= alertLevel) {
                        AlertReason reason = new AlertReason();
                        if (data[CREATE_TM_IDX] != null) {
                            reason.setTime((String) data[CREATE_TM_IDX]);
                        }
                        if (data[CLIENT_IP_IDX] != null) {
                            reason.setClientIp((String) data[CLIENT_IP_IDX]);
                        }
                        if (data[SRC_AUDIT_OBJ_IDX] != null) {
                            reason.setAuditObjId((String) data[SRC_AUDIT_OBJ_IDX]);
                        }
                        if (data[USR_IDX] != null) {
                            reason.setClientUsr((String) data[USR_IDX]);
                        }
                        if (data[SRC_DB_NAME_IDX] != null) {
                            reason.setDbName((String) data[SRC_DB_NAME_IDX]);
                        }
                        reason.setEventId(id);
                        if (data[SRC_LEVEL_IDX] != null) {
                            reason.setLevel((String) data[SRC_LEVEL_IDX]);
                        }
                        if (data[SRC_OPER_TYPE_IDX] != null) {
                            reason.setOpType((String) data[SRC_OPER_TYPE_IDX]);
                        }
                        if (data[SRC_RESRC_ID_IDX] != null) {
                            reason.setResrcId((String) data[SRC_RESRC_ID_IDX]);
                        }
                        if (data[TAB_NUM_ROWS_IDX] != null) {
                            reason.setRowNum((String) data[TAB_NUM_ROWS_IDX]);
                        }
                        if (data[SRC_RULE_ID_IDX] != null) {
                            reason.setRuleId((String) data[SRC_RULE_ID_IDX]);
                        }
                        if (data[SRC_TAB_NAME_IDX] != null) {
                            reason.setTblName((String) data[SRC_TAB_NAME_IDX]);
                        }
                        reason.setType("1");
                        reason.setLogType(id.substring(0, 5));
                        alertReasons.add(reason);
                    }

                    size++;
                }
                accumulator.add(size);
                if (alertReasons.size() > 0) {
                    sendAlert(alertReasons, rabbitMqConfBc.getValue());
                }
                return res.iterator();
            }).saveAsNewAPIHadoopDataset(hbaseJob.getConfiguration());
            this.dStream.afterBatchProcessed();
        });
    }

    @Override
    public void afterProcess(StreamWork.Context context) {
//        this.dStream.afterBatchProcessed();
    }

    private static void addCol(Object val, byte[] col, Put put) {
        if (val != null) {
            String str = String.valueOf(val);
            if (!str.isEmpty()) {
                put.addColumn(COL_FAMILY, col, Bytes.toBytes(str));
            }
        }
    }
}
